package study.rocketmq.origin;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;

public class Consumer2 {
    static Logger logger = LoggerFactory.getLogger(Consumer2.class);

    public static final String NAMESRC_ADDR = "120.25.73.46:9876";
    public static void main(String[] args) throws Exception {
        /**
         * TODO
         */
        //参数为 消费者组。
        DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer("Consumer-Group-02");

        defaultMQPushConsumer.setNamesrvAddr(NAMESRC_ADDR);
        defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        // 支持多个使用||分割。如 TagA||TagB
        // 如果send中有多个Tag，比如TagA||TagB，消费者只监听了一个TagA，那么TagB也可能被拉取到消费者中，但是会忽略TagB的消息（等于消息丢失了）。
        // 可以通过不同消费者组解决丢失问题。
        // 如果send比如TagA，TagB，消费者只监听了一个TagA，然后同一消费者组的启动TagB，那么TagA无效，TagB有效
        // 原因：同一个消费者组中，后订阅的Tag会覆盖先订阅的Tag。
        defaultMQPushConsumer.subscribe("topic-17","TagB");
        /**TODO 一次性拉取多少消息, 默认1  */
        defaultMQPushConsumer.setConsumeMessageBatchMaxSize(20);


        /**TODO 最大线程数和最小线程数  默认20  */
        defaultMQPushConsumer.setConsumeThreadMax(20);
        defaultMQPushConsumer.setConsumeThreadMin(20);

        defaultMQPushConsumer.setMaxReconsumeTimes(1);

        defaultMQPushConsumer.setMessageModel(MessageModel.CLUSTERING);


        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {

            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                MessageExt me = msgs.get(0);
                int QueueId = context.getMessageQueue().getQueueId();
                try {
                    String topic = me.getTopic();
                    String tags = me.getTags();
                    String keys = me.getKeys();
                    logger.info(keys+"-开始消费 --msgs.size ="+msgs.size() );
					if(keys.equals("key-10")) {
						int a = 1/0;
					}

                    String msgBody = new String(me.getBody(), RemotingHelper.DEFAULT_CHARSET);
                    logger.info(QueueId+"-消费成功- topic: " + topic + ",tags: " + tags + ", keys: " + keys + ",body: " + msgBody);
                } catch (Exception e) {
                    e.printStackTrace();
                    /** TODO 重试的次数（不包括第一次失败） */
					int recousumeTimes = me.getReconsumeTimes();
                    /** TODO 重发3次就不重试了 */
					if(recousumeTimes == 3) {
						//		记录日志....
						//  	做补偿处理
                        /** TODO 设置返回成功，就不重试了。 */
                        logger.error("消费失败-已经重试了【"+recousumeTimes+"】次，不再重试了 -- >" );
						return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
					}else{
                        logger.error("消费失败-准备重试- 已重试次数: " + recousumeTimes);
                    }
                    /** TODO RabbitMQ会自动重试，共1次，重试时间间隔如下：
                     *     10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
                     *     ps：延迟消息多了两档：1s 5s，其余一样。
                     *         messageDelayLevel = 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
                     * */
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        defaultMQPushConsumer.start();
        System.err.println("consumer start...");
    }
}
